package com.xazenith.kfk.config;

import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Objects;
import java.util.Properties;

/**
 * 配置信息
 *
 * @author zhouwei
 */
public class Configuration {

    private Configuration() {
    }

    /**
     * 启动服务器配置
     */
    public static final String KFK_CONFIG_BOOTSTRAP_SERVERS = "bootstrap.servers";


    /**
     * 消费者组Id的配置项
     */
    public static final String KFK_CONFIG_GROUP_ID = "group.id";

    /**
     * 消费者Id的配置项
     */
    public static final String KFK_CONFIG_CLIENT_ID = "client.id";

    /**
     * 消费者自动提交处理的配置项
     */
    public static final String KFK_CONFIG_ENABLE_AUTO_COMMIT = "enable.auto.commit";

    /**
     * key序列化配置项
     */
    public static final String KFK_CONFIG_KEY_SERIALIZER = "key.serializer";

    /**
     * value序列化配置项
     */
    public static final String KFK_CONFIG_VALUE_SERIALIZER = "value.serializer";

    /**
     * key序列化配置项
     */
    public static final String KFK_CONFIG_KEY_DESERIALIZER = "key.deserializer";

    /**
     * value序列化配置项
     */
    public static final String KFK_CONFIG_VALUE_DESERIALIZER = "value.deserializer";

    /**
     * 消费者默认情况下拉取消息等待的时间
     */
    public static final int KFK_CONFIG_DEFAULT_POLL_IN_MILLS = 100;

    /**
     * acks配置
     * 该值被默认配置为  all
     */
    public static final String ACKS = "acks";


    /**
     * 消费者配置项
     *
     * @param bootstrapServers kafka服务器配置
     * @param groupId 分组Id配置
     * @param autoCommit 是否允许自动提交
     *
     * @return {@link Properties}
     */
    public static Properties consumerConfig(String bootstrapServers, String groupId, boolean autoCommit) {
        Properties properties = new Properties();
        properties.setProperty(KFK_CONFIG_BOOTSTRAP_SERVERS, bootstrapServers);
        properties.setProperty(KFK_CONFIG_GROUP_ID, groupId);
        properties.setProperty(KFK_CONFIG_KEY_DESERIALIZER, StringDeserializer.class.getName());
        properties.setProperty(KFK_CONFIG_VALUE_DESERIALIZER, StringDeserializer.class.getName());
        properties.setProperty(KFK_CONFIG_ENABLE_AUTO_COMMIT, String.valueOf(autoCommit));
        return properties;
    }


    /**
     * 获取生产者配置的调用方法, acks为 'all'
     *
     * @param bootstrapServers kafka服务器配置
     *
     * @return {@link Properties}
     *
     * @see ACKSType
     */
    public static Properties producerConfig(String bootstrapServers) {
       return producerConfig(bootstrapServers, null);
    }


    /**
     * 获取生产者配置的调用方法
     *
     * @param bootstrapServers kafka服务器配置
     *
     * @param acksType acks类型
     *
     * @return {@link Properties}
     *
     * @see ACKSType
     */
    public static Properties producerConfig(String bootstrapServers, ACKSType acksType) {
        Properties properties = new Properties();
        properties.setProperty(KFK_CONFIG_BOOTSTRAP_SERVERS, bootstrapServers);
        properties.setProperty(KFK_CONFIG_KEY_SERIALIZER, StringSerializer.class.getName());
        properties.setProperty(KFK_CONFIG_VALUE_SERIALIZER, StringSerializer.class.getName());
        if (Objects.nonNull(acksType)) {
            properties.setProperty(ACKS, acksType.getTag());
        } else {
            properties.setProperty(ACKS, "all");
        }
        return properties;
    }


    /**
     * acks配置 enum
     */
    public enum ACKSType {
        /**
         * 0: 生产者在成功写入消息之前不会等待任何来自服务器的响应。也就是说，如果当中出现了问题，导致服务器没有收到消息，那么生产者就无从得知，
         * 消息也就丢失了。不过，因为生产者不需要等待服务器的响应，所以它可以以网络能够支持的最大速度发送消息，从而达到很高的吞吐量。
         */
        NONE("0"),

        /*
         * 1: 只要集群的首领节点收到消息，生产者就会收到一个来自服务器的成功响应。如果消息无法到达首领节点（比如首领节点崩溃，
         * 新的首领还没有被选举出来），生产者会收到一个错误响应，为了避免数据丢失，生产者会重发消息。如果一个没有收到消息的节点成为新首领，
         * 消息还是会丢失。这个时候的吞吐量取决于使用的是同步发送还是异步发送。如果让发送客户端等待服务器的响应（通过调用 Future 对象的 get（）
         * 方法），显然会增加延迟（在网络上传输一个来回的延迟）。如果客户端使用回调，延迟问题就可以得到缓解，不过吞吐量还是会受发送中消息数量的限制
         * （比如，生产者在收到服务器响应之前可以发送多少个消息）
         */
        WAIT_HEAD("1"),

        /*
         * all: 只有当所有参与复制的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应。这种模式是最安全的，
         * 它可以保证不止一个服务器收到消息，就算有服务器发生崩溃，整个集群仍然可以运行。它的延迟比 acks=1 时更高。
         */
        WAIT_ALL("all");

        private final String tag;

        ACKSType(String tag) {
            this.tag = tag;
        }

        public String getTag() {
            return tag;
        }
    }


}
